package org.apache.lucene.ocean;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.lucene.index.IndexCommit;
import org.apache.lucene.index.IndexDeletionPolicy;
import org.apache.lucene.index.IndexReader;
import org.apache.lucene.index.OceanSegmentReader;
import org.apache.lucene.ocean.Batch.SlaveBatch;
import org.apache.lucene.ocean.util.Constants;
import org.apache.lucene.ocean.util.SortedListMap;
import org.apache.lucene.ocean.util.Util;
import org.apache.lucene.store.Directory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Abstract class used by RamIndex and DiskIndex. Assumes a
 * org.apache.lucene.store.Directory based IndexReader implementation.
 * 
 */
public abstract class DirectoryIndex extends Index {
  final static Logger LOG = LoggerFactory.getLogger(DirectoryIndex.class);
  protected final SortedListMap<Long,DirectoryIndexSnapshot> indexSnapshotMap = new SortedListMap<Long,DirectoryIndexSnapshot>();
  private int numDeleteTransactionsSinceLastFlush = 0;
  protected DirectoryIndexDeletionPolicy indexDeletionPolicy = new DirectoryIndexDeletionPolicy();
  protected IndexReader initialIndexReader;
  protected final List<Long> checkpointSnapshotIds = new ArrayList<Long>();
  protected final List<Long> lastAppliedSnapshotIds = new ArrayList<Long>();
  private Long maxSnapshotId;
  private Long maxDocumentId;
  private Long minSnapshotId;
  private Long minDocumentId;

  public DirectoryIndex(IndexID id, TransactionSystem system) {
    super(id, system);
  }

  public void close() throws IOException {
    for (DirectoryIndexSnapshot snapshot : indexSnapshotMap.values()) {
      snapshot.close();
    }
  }

  public DirectoryIndexSnapshot getLatestIndexSnapshot() {
    return indexSnapshotMap.lastValue();
  }

  public DirectoryIndexSnapshot getIndexSnapshot(Long snapshotId) {
    return indexSnapshotMap.get(snapshotId);
  }

  /**
   * public IndexSnapshot initialize(Long snapshotId, List<Deletes>
   * deletesList, TransactionSystem system) throws Exception, IndexException,
   * IOException { IndexReader indexReader = initialIndexReader; if (deletesList ==
   * null || deletesList.size() == 0) { createNewSnapshot(snapshotId, false,
   * indexReader); } else { for (Deletes deletes : deletesList) {
   * applyDeletes(true, deletes, null, indexReader); indexReader.flush(); //
   * always flush here because it is the during the creation of the index
   * createNewSnapshot(slaveBatch.getId(), true, indexReader); } } assert
   * snapshotId.equals(indexSnapshotMap.lastKey()); return
   * indexSnapshotMap.get(indexSnapshotMap.lastKey()); }
   */

  public IndexSnapshot initialize(Long snapshotId, List<SlaveBatch> deleteOnlySlaveBatches, TransactionSystem system) throws Exception,
      IndexException, IOException {
    IndexReader indexReader = initialIndexReader;
    if (deleteOnlySlaveBatches == null || deleteOnlySlaveBatches.size() == 0) {
      createNewSnapshot(snapshotId, true, indexReader);
    } else {
      int numDeleted = 0;
      boolean checkpoint = false;
      for (int x = 0; x < deleteOnlySlaveBatches.size(); x++) {
        SlaveBatch slaveBatch = deleteOnlySlaveBatches.get(x);
        if (slaveBatch.hasDeletes()) {
          IndexSnapshotCommitResult deletesResult = applyDeletes(true, slaveBatch.getDeletes(), null, indexReader);
          numDeleted += deletesResult.getNumDeleted();
        }
        if (numDeleted > 0 && x == deleteOnlySlaveBatches.size() - 1) {
          checkpoint = true;
          indexReader.flush();
        }
        createNewSnapshot(slaveBatch.getId(), checkpoint, indexReader);
      }
    }
    // assert snapshotId.equals(indexSnapshotMap.lastKey());
    return indexSnapshotMap.lastValue();
  }

  protected void onCommit() throws Exception {

  }

  private List<DirectoryIndexSnapshot> getSnapshotsByGeneration(long generation) throws IOException {
    List<DirectoryIndexSnapshot> snapshots = new ArrayList<DirectoryIndexSnapshot>();
    for (DirectoryIndexSnapshot indexSnapshot : indexSnapshotMap.values()) {
      if (indexSnapshot.getIndexReader().getIndexCommit().getGeneration() == generation) {
        snapshots.add(indexSnapshot);
      }
    }
    return snapshots;
  }

  /**
   * Finds reader by version by iterating over snapshots and comparing versions
   * 
   * @param version
   * @return
   */
  private DirectoryIndexSnapshot getSnapshotByReaderVersion(long version) {
    for (DirectoryIndexSnapshot indexSnapshot : indexSnapshotMap.values()) {
      if (indexSnapshot.getIndexReaderVersion() == version) {
        return indexSnapshot;
      }
    }
    return null;
  }

  public class DirectoryIndexDeletionPolicy implements IndexDeletionPolicy {
    private IndexCommit lastCommit;
    SortedListMap<Long,IndexCommit> commitPoints = new SortedListMap<Long,IndexCommit>(); // key

    // is
    // generation

    // is
    // generation

    public void onInit(List commits) throws IOException {
      onCommit(commits);
    }

    public IndexCommit getLastIndexCommitPoint() {
      return lastCommit;
    }

    public void onCommit(List commits) throws IOException {
      try {
        commitPoints.clear();
        for (int x = 0; x < commits.size(); x++) {
          IndexCommit indexCommit = (IndexCommit) commits.get(x);
          commitPoints.put(indexCommit.getGeneration(), indexCommit);
        }
        DirectoryIndex.this.onCommit();
        lastCommit = (IndexCommit) commits.get(commits.size() - 1);
        for (int x = 0; x < commits.size() - 1; x++) {
          IndexCommit indexCommitPoint = (IndexCommit) commits.get(x);
          // Multiple snapshots may have the same generation
          // so deleting a commitpoint may affect multiple snapshots
          long generation = indexCommitPoint.getGeneration();
          List<DirectoryIndexSnapshot> snapshots = getSnapshotsByGeneration(generation);
          // if there are no snapshots it needs to be deleted, nothing
          // is using it anymore
          if (snapshots.size() == 0) {
            indexCommitPoint.delete();
            commitPoints.remove(indexCommitPoint.getGeneration());
          }
          for (DirectoryIndexSnapshot indexSnapshot : snapshots) {
            if (!indexSnapshot.hasRef()) {
              // not referenced in Snapshots anymore
              indexCommitPoint.delete();
              indexSnapshot.delete();
            }
          }
        }
      } catch (Exception exception) {
        throw Util.asIOException(exception);
      }
    }
  }

  public abstract class DirectoryIndexSnapshot extends IndexSnapshot {
    protected OceanSegmentReader indexReader;
    private int maxDoc;

    public DirectoryIndexSnapshot(Long snapshotId, OceanSegmentReader indexReader) throws IOException {
      super(snapshotId);
      this.indexReader = indexReader;
      maxDoc = indexReader.maxDoc();
      getMinDocumentId();
      getMaxDocumentId();
    }

    public void delete() throws Exception {
      // TODO: maybe need to check for snapshot refs here
      LOG.info(DirectoryIndex.this.getId() + " deleting snapshotid: " + snapshotId);
      long generation = getGeneration();
      List<DirectoryIndexSnapshot> snapshotsGreaterWithGeneration = new ArrayList<DirectoryIndexSnapshot>();
      for (DirectoryIndexSnapshot snapshot : indexSnapshotMap.values()) {
        if (snapshot.getGeneration() == generation && snapshot.snapshotId.longValue() > snapshotId) {
          snapshotsGreaterWithGeneration.add(snapshot);
        }
      }
      indexReader.close(false);
      if (snapshotsGreaterWithGeneration.size() == 0) {
        IndexCommit indexCommit = indexDeletionPolicy.commitPoints.get(generation);
        if (indexCommit != null) {
          LOG.info(DirectoryIndex.this.getId() + " deleting snapshotid: " + snapshotId + " indexCommit: " + indexCommit.getGeneration());
          indexCommit.delete();
        }
      }
      indexSnapshotMap.remove(snapshotId);
    }

    public Long getMinDocumentId() throws IOException {
      if (minDocumentId == null) {
        String string = getMin(Constants.DOCUMENTID);
        if (string == null)
          return null;
        minDocumentId = Util.longFromEncoded(string);
      }
      return minDocumentId;
    }

    public Long getMinSnapshotId() throws IOException {
      if (minSnapshotId == null) {
        String string = getMin(Constants.SNAPSHOTID);
        if (string == null)
          return null;
        minSnapshotId = Util.longFromEncoded(string);
      }
      return minSnapshotId;
    }

    public Long getMaxSnapshotId() throws IOException {
      if (maxSnapshotId == null) {
        String string = getMax(Constants.SNAPSHOTID);
        if (string == null)
          return null;
        maxSnapshotId = Util.longFromEncoded(string);
      }
      return maxSnapshotId;
    }

    public Long getMaxDocumentId() throws IOException {
      if (maxDocumentId == null) {
        String string = getMax(Constants.DOCUMENTID);
        if (string == null)
          return null;
        maxDocumentId = Util.longFromEncoded(string);
      }
      return maxDocumentId;
    }

    public Long getLastAppliedSnapshotId() {
      return getLastId(lastAppliedSnapshotIds, "getLastAppliedSnapshotId");
    }

    private Long getLastId(List<Long> ids, String name) {
      if (ids.size() == 0)
        return null;
      return ids.get(ids.size() - 1);
      /**
       * for (int x = ids.size() - 1; x >= 0; x--) { Long id = ids.get(x); if
       * (id.longValue() <= getSnapshotId()) { return id; } } Long lastId =
       * ids.get(ids.size() - 1); LOG.error(name + " pos is negative, should not
       * be possible for: " + getSnapshotId() + " last valid id: " + lastId);
       * return null;
       */
    }

    public Long getLastCheckpointId() {
      return getLastId(checkpointSnapshotIds, "getLastSnapshotIdDeletesFlushed");
    }

    public long getGeneration() throws IOException {
      return indexReader.getIndexCommit().getGeneration();
    }

    public void close() throws IOException {
      indexReader.close(false);
      indexSnapshotMap.remove(snapshotId);
    }

    public int deletedDoc() {
      return indexReader.maxDoc() - indexReader.numDocs();
    }

    public int maxDoc() {
      return indexReader.maxDoc();
    }

    public IndexReader getIndexReader() {
      return indexReader;
    }

    public long getIndexReaderVersion() {
      return indexReader.getVersion();
    }

    public boolean hasRef() throws Exception {
      return getSystem().getSnapshots().hasRefs(snapshotId);
    }
  }

  public long getSize() throws IOException {
    Directory directory = getDirectory();
    return Util.getSize(directory);
  }

  /**
   * Creates a new snapshot only without performing any changes to the index
   * 
   * @param transaction
   * @throws IndexException
   * @throws InterruptedException
   * @throws IOException
   */
  public void commitNothing(Transaction transaction) throws IndexException, InterruptedException, IOException, Exception {
    IndexSnapshot latestIndexSnapshot = getLatestIndexSnapshot();
    assert latestIndexSnapshot != null;
    //if (latestIndexSnapshot.getSnapshotId().equals(transaction.getPreviousId())) {
    //}
    transaction.ready(this);
    if (transaction.go()) {
      Long snapshotId = transaction.getId();
      IndexReader previousIndexReader = latestIndexSnapshot.getIndexReader();
      IndexReader newIndexReader = (IndexReader)previousIndexReader.clone();
      createNewSnapshot(snapshotId, false, newIndexReader);
      removeOldSnapshots(indexSnapshotMap);
    }
  }

  public boolean rollback(Long snapshotId) throws Exception {
    LOG.info("rollback " + snapshotId);
    DirectoryIndexSnapshot indexSnapshot = indexSnapshotMap.get(snapshotId);
    if (indexSnapshot != null) {
      indexSnapshot.delete();
      return true;
    }
    return false;
  }

  public IndexSnapshotCommitResult commitDeletes(Deletes deletes, Transaction transaction) throws Exception, IndexException, InterruptedException,
      IOException {
    IndexSnapshot latestIndexSnapshot = getLatestIndexSnapshot();
    assert latestIndexSnapshot != null;
    // assert
    // latestIndexSnapshot.getSnapshotId().equals(transaction.getPreviousId());
    // if
    // (latestIndexSnapshot.getSnapshotId().equals(transaction.getPreviousId()))
    // {
    // System.out.println("latest: " + latestIndexSnapshot.getSnapshotId() + "
    // previous: " + transaction.getPreviousId());
    // }
    IndexReader previousIndexReader = latestIndexSnapshot.getIndexReader();
    IndexReader newIndexReader = (IndexReader) previousIndexReader.clone();
    try {
      IndexSnapshotCommitResult deletesResult = applyDeletes(true, deletes, null, newIndexReader);
      transaction.ready(this);
      if (transaction.go()) {
        // if (deletesResult.getNumDeleted() > 0) {
        // if (TransactionSystem.flushIndexReaderAfterDelete())
        // newIndexReader.flush();
        // }
        boolean checkpoint = false;
        int deleteThreshold = (int) (getSystem().getConfig().deletesFlushThresholdPercent * newIndexReader.maxDoc());
        if (deleteThreshold > 0 && deletesResult.getNumDeleted() > 0 && numDeleteTransactionsSinceLastFlush > deleteThreshold) {
          if (LOG.isInfoEnabled())
            LOG.info("deletes flushed numDeleteTransactionsSinceLastFlush: " + numDeleteTransactionsSinceLastFlush + " threshold: "
                + deleteThreshold);
          newIndexReader.flush();
          checkpoint = true;
        }
        Long snapshotId = transaction.getId();
        if (deletesResult.getNumDeleted() > 0) {
          numDeleteTransactionsSinceLastFlush++;
          lastAppliedSnapshotIds.add(snapshotId);
          // if flush was called check to make sure there is a new generation
          // for the indexreader
          LOG.info("previous reader gen: " + previousIndexReader.getIndexCommit().getGeneration() + " newIndexReader gen: "
              + newIndexReader.getIndexCommit().getGeneration());
        }
        createNewSnapshot(snapshotId, checkpoint, newIndexReader);
        removeOldSnapshots(indexSnapshotMap);
        return deletesResult;
      } else {
        rollback(transaction.getId());
        return null;
      }
    } catch (Throwable throwable) {
      LOG.error("error in " + DirectoryIndex.this.getId(), throwable);
      transaction.failed(this, throwable);
      // rollback(transaction.getId());
      return null;
    }
  }

  protected DirectoryIndexSnapshot createNewSnapshot(Long snapshotId, boolean checkpoint, IndexReader newIndexReader) throws IOException {
    if (newIndexReader == null)
      newIndexReader = this.initialIndexReader;
    if (checkpoint) {
      assert (newIndexReader.maxDoc() - newIndexReader.numDocs()) >= 0;
      numDeleteTransactionsSinceLastFlush = 0;
      checkpointSnapshotIds.add(snapshotId);
    }
    return doCreateNewSnapshot(snapshotId, newIndexReader);
  }

  protected abstract DirectoryIndexSnapshot doCreateNewSnapshot(Long snapshotId, IndexReader newIndexReader) throws IOException;

  protected void registerSnapshot(DirectoryIndexSnapshot indexSnapshot) throws IOException {
    indexSnapshotMap.put(indexSnapshot.getSnapshotId(), indexSnapshot);
  }

  public abstract Directory getDirectory();
}
